package com.hp.bon.sgw.service;

import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hp.bon.sgw.Constants;
import com.hp.bon.sgw.Constants.NodeType;
import com.hp.bon.sgw.bean.CapabilityCtrlBean;
import com.hp.bon.sgw.bean.FluxSumBean;
import com.hp.bon.sgw.bean.SysParamBean;
import com.hp.bon.sgw.bean.XmlMessage;
import com.hp.bon.sgw.convert.Xml2AVPConvetor;
import com.hp.bon.sgw.core.CallParam;
import com.hp.bon.sgw.domain.ConvMapInfo;
import com.hp.bon.sgw.domain.Node;
import com.hp.bon.sgw.domain.OpenedCapability;
import com.hp.bon.sgw.domain.Top100KeyWordBean;
import com.hp.bon.sgw.start.SpringContext;
import com.hp.bon.sgw.threads.Top100CacheManager;
import com.hp.bon.sgw.util.Comutils;
import com.hp.bon.sgw.util.dcc.DccAvpHelper;
import com.hp.ocs.diameter.message.Message;

/**
 * 流量控制服务，计算流量，确定是否断连
 * 
 * @author wuzhih
 * 
 */
public class FluxControlService implements ConfigListener {

	private static FluxControlService instance;

	public static FluxControlService getInstance() {
		if (instance == null) {
			instance = new FluxControlService();
			instance.init();

			// Add by ZT:20130629 13:01:39
			SpringContext.getInstance().getConfigService().addConfigListener(instance);
		}
		return instance;
	}

	/**
	 * Logger for this class
	 */
	private static final Logger logger = LoggerFactory.getLogger(FluxControlService.class);

	/**
	 * 所关心的运行期变化的配置数据的Key
	 */
	private final Set<String> monitorConfKeys = new HashSet<String>();
	private String fluxContrlDeniniedPeriodDccKey = "business.fluxctrl.dcc.deniedseconds";
	private String fluxContrlDeniniedPeriodSoapKey = "business.fluxctrl.soap.deniedseconds";
	private String fluxContrlDeniniedPeriodScspKey = "business.fluxctrl.scsp.deniedseconds";
	private String autoFluxCtrolCalcPeriodKey = "business.fluxctrl.auto.calcperiod";
	private String autofluxCtrlCalcFomulaKey = "business.fluxctrl.auto.calcformula";
	private String autoFluxCtrolDeniniedTopKey = "business.fluxctrl.auto.deniniedtop10";
	private String autoFluxCtrolCongestionThresHoldKey = "business.fluxctrl.auto.congestion-threshold";
	private String fluxCtrlPolicyKey = "business.fluxctrl.autocontrol";

	private boolean filterContent = false;
	private volatile int fluxContrlDeniniedSencondsDcc = 300;
	private volatile int fluxContrlDeniniedSencondsSoap = 300;
	private volatile int fluxContrlDeniniedSencondsScsp = 300;

	public FluxControlService() { 
		monitorConfKeys.add(SysParamBean.KEY_BUSINESS_If_filter_Content);
		monitorConfKeys.add(fluxContrlDeniniedPeriodDccKey);
		monitorConfKeys.add(fluxContrlDeniniedPeriodSoapKey);
		monitorConfKeys.add(fluxContrlDeniniedPeriodScspKey);
		monitorConfKeys.add(autoFluxCtrolCalcPeriodKey);
		monitorConfKeys.add(this.fluxCtrlPolicyKey);
		monitorConfKeys.add(autofluxCtrlCalcFomulaKey);
		monitorConfKeys.add(autoFluxCtrolDeniniedTopKey);
		monitorConfKeys.add(this.autoFluxCtrolCongestionThresHoldKey);
	}

	/**
	 * 记录流量，key是capability+nehost+user
	 */
	private volatile Map<String, FluxSumBean> fluxSumMap = Collections.emptyMap();
	/**
	 * 因为流量超标而被临时禁止的网元，key是网元hostID,value是禁止截至期限（时间戳）
	 */
	private volatile Map<String, Long> denniedNEMap = Collections.emptyMap();

	public Map<String, Long> getDenniedNEMap() {
		return denniedNEMap;
	}

	public boolean checkAllowedConnect(String neHost) {
		Long denniedTime = denniedNEMap.get(neHost);
		if (denniedTime == null) {
			return true;
		} else {
			// System.out.println(Comutils.formatTime(denniedTime,
			// "yyyy-MM-dd HH:mm:ss.SSS"));
			if (denniedTime < System.currentTimeMillis()) {
				// 已经过期
				return true;
			} else {
				return false;
			}
		}
	}

	public void init() {
		loadParams();
	}

	public void doPeriodWork() {
		checkAndRestFluxBean();
	}

	/**
	 * 检查用户流量是否超标
	 * 
	 * @param capability
	 * @param data
	 *            .getNeHost()
	 * @param user
	 * @param msgLength
	 * @return
	 */
	public int checkUserFlux(CallParam callParam, int msgLength) {
		int retCode = 0;// 成功
		Node node = callParam.getFromNode();
		String neHost = callParam.getUserName();
		if (!this.checkAllowedConnect(neHost+callParam.getCapability())) {
			logger.warn(Constants.ResultCode.FLUX_DISCONNECT + "|" + neHost + " has been dennied,please wait some more minutes ");
			return Constants.ResultCode.FLUX_DISCONNECT;
		}

		CapabilityCtrlBean capabilityBean = SpringContext.getInstance().getConfigService().getCapabilityCtrlMap().get(callParam.getCapability());
		// 检查包大小
		if (capabilityBean != null && capabilityBean.getMaxReqSize() > -1 && msgLength > capabilityBean.getMaxReqSize()) {
			logger.warn("capability :" + callParam.getCapability() + " from node " + neHost + " sended request message size too large " + msgLength + " threshold:" + capabilityBean.getMaxReqSize());
			retCode = Constants.ResultCode.MESSAGE_SIZE_OVER;
			return retCode;
		}

		if (callParam.getCapabilityBean() == null || callParam.getCapabilityBean().getControlType() == OpenedCapability.CONTROL_NONE) {
			// 允许通过
			return 0;
		}

		String reason = "";
		FluxSumBean fluxBean = findFluxSumBean(callParam.getCapability() + callParam.getUserName());
		fluxBean.increAccess(msgLength);
		int totalReq = fluxBean.getTotalRequests();
		int totalBytes = fluxBean.getTotalBytes();

		int deniniedSed = this.fluxContrlDeniniedSencondsDcc;
		switch (node.getNodeType()) {
		case NodeType.NODE_WS_NE:
			deniniedSed = this.fluxContrlDeniniedSencondsSoap;
		case NodeType.NODE_SCSP_NE:
			deniniedSed = this.fluxContrlDeniniedSencondsScsp;
		default:
			deniniedSed = this.fluxContrlDeniniedSencondsDcc;
		}

		switch (callParam.getCapabilityBean().getControlType()) {
		case OpenedCapability.CONTROL_BOTH: {
			if (totalBytes > callParam.getCapabilityBean().getFluxDisconnect() * 1000 || totalReq > callParam.getCapabilityBean().getCountDisconnect()) {
				retCode = Constants.ResultCode.FLUX_DISCONNECT;
				putToDenined(node.getHostId()+callParam.getCapability(), deniniedSed);
				reason = "flux exceed disconnect threshold values";
			} else if (totalBytes > callParam.getCapabilityBean().getFluxWarning() * 1000 || totalReq > callParam.getCapabilityBean().getCountWarning()) {
				retCode = Constants.ResultCode.FLUX_WARNING;
				reason = "flux exceed warning threshold values";
			}
			break;
		}
		case OpenedCapability.CONTROL_BY_COUNT: {
			if (totalReq > callParam.getCapabilityBean().getCountDisconnect()) {
				retCode = Constants.ResultCode.FLUX_DISCONNECT;
				putToDenined(node.getHostId(), deniniedSed);
				reason = "count exceed disconnect threshold values";
			} else if (totalReq > callParam.getCapabilityBean().getCountWarning()) {
				retCode = Constants.ResultCode.FLUX_WARNING;
				reason = "count exceed warning threshold values";
			}
			break;
		}
		case OpenedCapability.CONTROL_BY_FLUX: {
			if (totalBytes > callParam.getCapabilityBean().getFluxDisconnect() * 1000) {
				retCode = Constants.ResultCode.FLUX_DISCONNECT;
				putToDenined(neHost, deniniedSed);
				reason = "size exceed disconnect threshold values";
			} else if (totalBytes > callParam.getCapabilityBean().getFluxWarning() * 1000) {
				retCode = Constants.ResultCode.FLUX_WARNING;
				reason = "size exceed warning threshold values";
			}
			break;
		}
		}
		if (retCode != 0) {
			if (retCode == Constants.ResultCode.FLUX_DISCONNECT || (retCode == Constants.ResultCode.FLUX_WARNING && (totalReq - callParam.getCapabilityBean().getCountWarning() == 1))) {
				logger.warn(Constants.SYSLOG_FluxContrlWarn + "|" + retCode + "|" + node.getNodeCode() + "|" + node.getHostId() + "|" + callParam.getCapability() + "|" + totalReq + "|" + totalBytes + "|" + node.getHostId() + " " + reason);
			} else {
				logger.warn(retCode + "|" + node.getHostId() + " request " + reason);
			}
		}
		return retCode;
	}

	/**
	 * 关键字统计
	 */
	public String filterContent(String serviceCtxID, int gateWayType, Object reqData) {
		CapabilityCtrlBean capCtrlBean = SpringContext.getInstance().getConfigService().getCapabilityCtrlMap().get(DccAvpHelper.getCapability(serviceCtxID));
		if (capCtrlBean == null) {
			return null;
		}
		Set<String> monitoryKeys = capCtrlBean.getKeyTagSet();
		boolean doWatchKey = (!capCtrlBean.isIfKeyStat()) || ((monitoryKeys != null) && (!monitoryKeys.isEmpty()));
		boolean doFilterContent = (this.filterContent) && (capCtrlBean.needFilterContent());
		if ((!doWatchKey) && (!doFilterContent)) {
			return null;
		}

		Set<String> allFields = new LinkedHashSet<String>();
		Map<String, String> keyWordMap = capCtrlBean.getFilterMap();
		if (doWatchKey) {
			allFields.addAll(monitoryKeys);
		}
		if (doFilterContent) {
			allFields.addAll(keyWordMap.keySet());
		}
		if (logger.isDebugEnabled()) {
			logger.debug("doFilterContent " + doFilterContent + " monitoryKeys and filter keys total " + Arrays.toString(allFields.toArray()) + " for " + serviceCtxID);
		}

		Map<String, String> fieldsMap = null;
		if (reqData instanceof XmlMessage) {
			fieldsMap = Comutils.getLeafNodeOfXMl(((XmlMessage) reqData).getDataElement(), allFields);
		} else if (reqData instanceof Message) {
			Map<String, ConvMapInfo> allAVPMap = Xml2AVPConvetor.getInstance().getALLNamesMap();
			fieldsMap = DccAvpHelper.converDCCMsgLeafField((Message) reqData, allFields, allAVPMap);
		} else {
			logger.warn("can't recoginize request data class " + reqData.getClass().getName());
			return null;
		}

		return filterContent(serviceCtxID, gateWayType, monitoryKeys, keyWordMap, fieldsMap, doFilterContent, doWatchKey);
	}

	private String filterContent(String serviceCtxID, int gateWayType, Set<String> monitoryKeys, Map<String, String> keyWordMap, Map<String, String> fieldsMap, boolean doFilterContent, boolean doWatchKey) {
		for (Map.Entry<String, String> pair : fieldsMap.entrySet()) {
			if (doFilterContent) {
				String keyword = (String) keyWordMap.get(pair.getKey());
				if ((keyword != null) && (((String) pair.getValue()).contains(keyword))) {
					return keyword;
				}
			}

			if ((doWatchKey) && (monitoryKeys.contains(pair.getKey()))) {
				writeTop100Rec(serviceCtxID, gateWayType, (String) pair.getKey(), (String) pair.getValue());
			}
		}
		return null;
	}

	private void writeTop100Rec(String serviceCtxID, int gateWayType, String key, String value) {
		if (logger.isDebugEnabled()) {
			logger.debug("write top 100 rec " + serviceCtxID + "  key " + key + " value " + value);
		}
		Top100KeyWordBean keyword = new Top100KeyWordBean();
		keyword.setserviceContextId(serviceCtxID);
		keyword.setProtocolType(gateWayType);
		keyword.setMonitorKey(key);
		keyword.setKeyContent(value);
		Top100CacheManager.getInstance().add(keyword);
		// SpringContext.getInstance().getDataStoreService().addJsonJava(keyword);
	}

	private void putToDenined(String neHostID, int denniedPeriod) {
		if (denniedPeriod == 0) {
			denniedPeriod = 300;
		}
		Long time = denniedNEMap.get(neHostID);
		Long newTime = System.currentTimeMillis() + denniedPeriod * 1000;
		if (time != null) {
			denniedNEMap.put(neHostID, newTime);
		} else {
			// 复制一份新的Map，并加入新限制的主机
			Map<String, Long> newMap = new LinkedHashMap<String, Long>(denniedNEMap);
			newMap.put(neHostID, newTime);
			this.denniedNEMap = newMap;
		}
		logger.info("denny " + neHostID + " until " + Comutils.formatTime(newTime, "yyyy-MM-dd HH:mm:ss.SSS"));
	}

	private FluxSumBean findFluxSumBean(String key) {
		FluxSumBean fluxBean = fluxSumMap.get(key);
		if (fluxBean == null) {
			Map<String, FluxSumBean> newFluxSumMap = new LinkedHashMap<String, FluxSumBean>(fluxSumMap);
			fluxBean = new FluxSumBean();
			newFluxSumMap.put(key, fluxBean);
			this.fluxSumMap = newFluxSumMap;
		}
		return fluxBean;
	}

	/**
	 * 周期性检查流控数据，进行清理
	 */
	private void checkAndRestFluxBean() {
		Map<String, CapabilityCtrlBean> crlBeanMap = SpringContext.getInstance().getConfigService().getCapabilityCtrlMap();
		for (Map.Entry<String, FluxSumBean> entry : this.fluxSumMap.entrySet()) {
			CapabilityCtrlBean ctrlBean = crlBeanMap.get(entry.getKey());
			if (ctrlBean != null && ctrlBean.getControlType() != OpenedCapability.CONTROL_NONE) {
				FluxSumBean fluBean = entry.getValue();
				if (fluBean.getResetTime() + ctrlBean.getFcontrolCycle() * 1000 < System.currentTimeMillis()) {
					fluBean.reset();
				}
			}
		}
	}

	public Set<String> monitorConfKeys() {
		return this.monitorConfKeys;
	}

	public void onConfigUpdated(String updatedKey) {
		loadParams();
	}

	private void loadParams() {
		int value = SpringContext.getInstance().getConfigService().getIntValue(SysParamBean.KEY_BUSINESS_If_filter_Content, 0);
		this.filterContent = (value == 1);
		logger.info("load param " + SysParamBean.KEY_BUSINESS_If_filter_Content + " to value " + value);
		value = SpringContext.getInstance().getConfigService().getIntValue(this.fluxContrlDeniniedPeriodDccKey, 300);
		logger.info("load param " + fluxContrlDeniniedPeriodDccKey + " to value " + value);
		this.fluxContrlDeniniedSencondsDcc = value;

		value = SpringContext.getInstance().getConfigService().getIntValue(this.fluxContrlDeniniedPeriodSoapKey, 300);
		logger.info("load param " + fluxContrlDeniniedPeriodSoapKey + " to value " + value);
		this.fluxContrlDeniniedSencondsSoap = value;

		value = SpringContext.getInstance().getConfigService().getIntValue(this.fluxContrlDeniniedPeriodScspKey, 300);
		logger.info("load param " + fluxContrlDeniniedPeriodScspKey + " to value " + value);
		this.fluxContrlDeniniedSencondsScsp = value;
	}

}